/* 
 * File:   JServerMaster.cpp
 * Author: johnneyshi
 * 
 * Created on 2013年1月27日, 下午8:16
 */

#include <signal.h>


#include "JServerMaster.h"
#include <arpa/inet.h>
#include <iostream>
#include <string>
#include <string.h>
#include <stdio.h>
#include "JException.h"
#include "JClient.h"
#include <fstream>
#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/xml_parser.hpp>
#include <boost/foreach.hpp>
#include "JLog.h"
#include "JDefine.h"
#include <sys/epoll.h>
#include "JEpoll.h"
#include <sys/syscall.h>
#define gettid() syscall(__NR_gettid)
using namespace std;



//std::queue<int> adapterConnFdQueue;
pthread_mutex_t JServerMaster::needDealFdsLock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t JServerMaster::needDealFdsCond = PTHREAD_COND_INITIALIZER;


JServerMaster::JServerMaster()
{
}

JServerMaster::~JServerMaster() 
{
}

void JServerMaster::Init(std::string _configFileName)
{
    ReadConfig(_configFileName);
}

void JServerMaster::Start()
{
    for(int i = 0;i < masterWorkerAdapters.size(); ++i)
    {
        AddWorker(masterWorkerAdapters[i]);
        JLog::Log(JLogMessage,"readPipe:%d\n",masterWorkerAdapters[i].readPipe);
        JServer::epoll.AddFd(masterWorkerAdapters[i].readPipe,EPOLLIN);
        //adapterFdsEpoll.AddFd(masterWorkerAdapters[i].readPipe,EPOLLIN);
    }
    JServer::Start();
}

void JServerMaster::AddWorker(MasterWorkerAdapter& adapter)
{
    JClient* pClient = adapter.pClient;
    WorkerArg* pWorkerArg = &(adapter.worker);
    JLog::Log(JLogMessage,"Add worker, client IP(%s) port(%d), worker IP(%s) port(%d)\n",
            pClient->GetClientIP().c_str(),pClient->GetClientPort(),pWorkerArg->ip.c_str(),pWorkerArg->port);
    pid_t childPid;
    int sockfd[2];
    socketpair(AF_LOCAL,SOCK_STREAM,0,sockfd);
    if((childPid = fork()) == 0)
    {
        try
        {
            JLog::Log(JLogMessage,"Create pid:%d\n",getpid());
            close(sockfd[0]);
            MasterWorkerAdapterMain(sockfd[1],sockfd[1],adapter);
        }
        catch(JException& e)
        {
            JLog::Log(JLogMessage,"%s\n",e.GetMessage().c_str());
        }
    }
    close(sockfd[1]);
    adapter.writePipe = sockfd[0];
    adapter.readPipe = sockfd[0];

}

void JServerMaster::ReadConfig(std::string _configFileName)
{
    using boost::property_tree::ptree;
    using namespace boost::property_tree::xml_parser;
    ptree pt;
    read_xml(_configFileName, pt);
    std::string masterIP = pt.get<std::string>("ServerMaster.IP");
    int masterPort = pt.get<int>("ServerMaster.Port");
    JServer::Init(masterIP,masterPort);
    
    BOOST_FOREACH(ptree::value_type &v,pt.get_child("ServerMaster.ServerWorkers"))
    {
        MasterWorkerAdapter adapter;
        std::string clientIP = v.second.get<std::string>("Client.IP","");
        int clientPort = v.second.get<int>("Client.Port",0);
        adapter.pClient = new JClient(clientIP,clientPort);
        
        adapter.worker.ip = v.second.get<std::string>("Server.IP","");
        adapter.worker.port = v.second.get<int>("Server.Port",0);
        adapter.worker.weight = v.second.get<int>("Weight");
        adapter.worker.threadCount = v.second.get<int>("ThreadCount");
        adapter.worker.threadMaxCount = v.second.get<int>("ThreadMaxCount");
        //adapter.pWorker = new JServerWorker;
        //adapter.pWorker->Init(ip,port,weight,adapter.threadCount,adapter.threadMaxCount);
        this->masterWorkerAdapters.push_back(adapter);
    }
}

//把接受到的消息，加上fd，然后转发给其中的一个worker
void JServerMaster::DealEpollFd(int fd)
{
    JLog::Log(JLogMessage,"From adapter read pipe:%d\n",fd);
    for(int i = 0;i < masterWorkerAdapters.size(); ++i)
    {
        if(masterWorkerAdapters[i].readPipe == fd)
        {
            int a;
            read(fd,(void*)&(a),SIZEOF_INT);
            JLog::Log(JLogMessage,"Index:%d --",i);
            --(masterWorkerAdapters[i].worker.currentWorkCount);
        }
    }
}

//计算公式：现在负载量=当前正在处理的工作/权重，选择现在负载量最小的进行分配
int JServerMaster::AssignWork()
{
    float minCapacity = numeric_limits<float>::max();
    int minIndex = -1;
    for(int i = 0;i < this->masterWorkerAdapters.size(); ++i)
    {
        
        float f = (float)(masterWorkerAdapters[i].worker.currentWorkCount) / masterWorkerAdapters[i].worker.weight;
        JLog::Log(JLogMessage,"currentWorkerCount:%d weight:%d",masterWorkerAdapters[i].worker.currentWorkCount,masterWorkerAdapters[i].worker.weight);
        printf(" f:%f\n",f);
        if(minCapacity > f)
        {
            minCapacity = f;
            minIndex = i;
        }
    }
    ++masterWorkerAdapters[minIndex].worker.currentWorkCount;
    JLog::Log(JLogMessage,"AssignWork:%d\n",minIndex);
    return minIndex;
}


void JServerMaster::DealNewConnfd(int newFd)
{
    //设置描述符为nonblock,为了防止拒绝攻击.
    SetNonBlock(newFd);
    int workIndex = AssignWork();
    JLog::Log(JLogMessage,"Master parent write new fd to adapter:%d\n",newFd);
    //把得到的描述符传给下标为workIndex的进程
    char c = ' ';
    WriteFd(masterWorkerAdapters[workIndex].writePipe,&c,1,newFd);
    //write(masterWorkerAdapters[workIndex].writePipe,(void*)&newFd,SIZEOF_INT);
    //write(newFd,(void*)&workIndex,sizeof(int));
    
    close(newFd);
}

/*Adapters是用于从队列中取出needDealFds,然后用缓存，得到所有的数据后，然后从ClientAdapters中选取一个
 * ClientAdapter，把数据发送给worker，然后接受来自worker处理完的数据，不需要缓存，直接传送给客户端，最后断开连接
 * 
 * 
 * 这里还需要一种机制：Adapter get new如果客户端拒绝攻击，或者发送的数据没有到size大小，那么就循环，从队列中另外再取一个描述符，进行处理
 * 但是原来的描述符，也用epoll监听着，最后只有接受到size大小的数据，才发往worker
 */
void* AdapterThreadMain(void* arg)
{
    //printf("address:%x\n",arg);
    //sleep(1);
    AdapterThreadArg* pArg = (AdapterThreadArg*)arg;
    JAdapterThread adapter(pArg);
    //JLog::Log(JLogMessage,"Pid:%d Port:%d\n",gettid(),pArg->port);
    adapter.Start();
}


/*
 * Adapters是用于从队列中取出needDealFds,然后用缓存，得到所有的数据后，然后从ClientAdapters中选取一个
 * ClientAdapter，把数据发送给worker。
 * ClientAdapter的用于发送数据给worker和接受来自worker的数据。
 */
void JServerMaster::MasterWorkerAdapterMain(int readFd,int writeFd,MasterWorkerAdapter& adapter)
{
    pthread_t tid;
    AdapterThreadArg* args = new AdapterThreadArg[adapter.worker.threadCount];/////////////这里没有delete
    //创建多线程池
    for(int i = 0;i < adapter.worker.threadCount; ++i)
    {
        AdapterThreadArg* pArg = &(args[i]);
        pArg->pAdapter = &adapter;
        pArg->pMaster = this;
        //printf("pMaster:%x,this:%x\n",pArg->pMaster,this);
        pArg->childReadPipe = readFd;
        pArg->childWritePipe = writeFd;
        pArg->port = adapter.pClient->GetClientPort() + i;
        //JLog::Log(JLogMessage,"pppp:%d\n",pArg->port);
        pthread_create(&tid,NULL,&AdapterThreadMain,(void*)pArg);//////////////////////////看看有没有数据共享方面的错误
    }
    //
    //pthread_create(&tid,NULL,&ListenRecvMsgWorker,(void*)this);
    //把接受到的要处理的connFd都放入一个队列中，各个线程从中取connFd;
    char c;
    while(true)
    {
        int newFd = -1;//
        //read(readFd,(void*)&newFd,sizeof(int));//////////////////////////////////////////
        if(ReadFd(readFd,&c,1,&newFd) == 0)
        {
            throw JException("ReadFd error.");
        }
        
        if(newFd < 0)
        {
            throw JException("ReadFd error, newFd < 0");
        }
        JLog::Log(JLogMessage,"Master child get new fd:%d\n",newFd);
        pthread_mutex_lock(&needDealFdsLock);
        needDealFds.push(newFd);
        //printf("address:%x\n",&needDealFds);
        pthread_cond_signal(&needDealFdsCond);
        pthread_mutex_unlock(&needDealFdsLock);
    }
}

